package demo;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WordCountTopology {

	private static final String SENTENCE_SPOUT_ID = "sentence_spout";

	private static final String SPLIT_BOLT_ID = "split_bolt";

	private static final String COUNT_BOLT_ID = "count_bolt";

	private static final String REPORT_BOLT_ID = "report_bolt";

	private static final String TOPOLOGY_NAME = "WordCountTopology";

	public static void main(String[] args) {
		SentenceSpout sentenceSpout = new SentenceSpout();
		SplitBolt splitBolt = new SplitBolt();
		CountBolt countBolt = new CountBolt();
		ReportBolt reportBolt = new ReportBolt();

		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout(SENTENCE_SPOUT_ID, sentenceSpout);

		builder.setBolt(SPLIT_BOLT_ID, splitBolt).shuffleGrouping(SENTENCE_SPOUT_ID);
		builder.setBolt(COUNT_BOLT_ID, countBolt,4)
//		.fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
		.shuffleGrouping(SPLIT_BOLT_ID);
		builder.setBolt(REPORT_BOLT_ID, reportBolt).globalGrouping(COUNT_BOLT_ID);

		Config config = new Config();

		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());

		Utils.waitForSeconds(5);

		cluster.killTopology(TOPOLOGY_NAME);
		cluster.shutdown();
	}
}
